package ar.edu.itba.pod.legajo45129.simul.communication;

import java.io.Serializable;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import ar.edu.itba.pod.legajo45129.Configuration;
import ar.edu.itba.pod.legajo45129.MessageProcessor;
import ar.edu.itba.pod.legajo45129.NodeId;
import ar.edu.itba.pod.legajo45129.simul.communication.factory.MessageFactory;
import ar.edu.itba.pod.simul.communication.AgentDescriptor;
import ar.edu.itba.pod.simul.communication.ClusterAdministration;
import ar.edu.itba.pod.simul.communication.ConnectionManager;
import ar.edu.itba.pod.simul.communication.Message;

import com.google.common.base.Preconditions;

public class ClusterAdministrationImpl implements ClusterAdministration, Serializable {
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	
	private String groupId;
	private Set<String> nodes;

	private ConnectionManagerImpl connectionManager;

//	private long TIMEOUT = 5000;
	private static int CONNECTION_FACTOR = 3;

	private static Logger logger = Logger.getLogger(ClusterAdministrationImpl.class);

	public ClusterAdministrationImpl(ConnectionManagerImpl connectionManager) throws RemoteException {
		Preconditions.checkNotNull(connectionManager);
		UnicastRemoteObject.exportObject(this, 0);
		this.connectionManager = connectionManager;
		this.nodes = new CopyOnWriteArraySet<String>();
	}

	@Override
	public Iterable<String> addNewNode(String newNode) throws RemoteException {
		if (!isConnectedToGroup()) {
			logger.error("node is not connected to a group");
			throw new IllegalStateException();
		}
		
		NodeId newNodeId = new NodeId(newNode);
		
		ConnectionManager cm;
		try {
			cm = connectionManager.getConnectionManager(newNode);
		} catch (Exception e) {
			logger.error("error getting connection manager of the remote node: " + e.getMessage());
			return nodes;
		}
		//check if group of newNode is the same to destination node group
		if (!this.groupId.equals(cm.getClusterAdmimnistration().getGroupId())) {
			logger.error("target node group is not equals to destination node group");
			throw new IllegalArgumentException();
		}
		
		// every node that added to the set is transformed to ip:port form in the DEBUG VERSION
		// otherwise, very node is saved in "ip" form
		
		this.nodes.add(newNodeId.toString());			
		
		return this.nodes;
	}

	@Override
	public void connectToGroup(String initialNode) throws RemoteException {
		NodeId initialNodeId = new NodeId(initialNode);
		logger.info("connecting to group through " + initialNodeId);
		ConnectionManager cm = connectionManager.getConnectionManager(initialNode);
		
		if (!isConnectedToGroup()) {
			// obtain the groupid of the entry point node, and set it as my groupid
			this.groupId = cm.getClusterAdmimnistration().getGroupId();			
		}
		
		if (this.nodes.contains(connectionManager.getNodeId().toString())) {
			logger.error("new node already exists in the group");
			throw new IllegalStateException("new node already exists in the group");
		}
		
		if (connectionManager.getNodeId().equals(initialNode)) {
			logger.error("new node equals initial node");
			throw new IllegalArgumentException("new node equals initial node");
		}
		
		if (nodes.size() < CONNECTION_FACTOR) {
			Iterator<String> it = cm.getClusterAdmimnistration().addNewNode(connectionManager.getNodeId().toString()).iterator();
			int count = 0;
			while (it.hasNext()) {
				NodeId node = new NodeId(it.next()); 
				if (!connectionManager.getNodeId().equals(node) && !this.nodes.contains(node)) {
					this.nodes.add(node.toString());		
					count++;
				}
				
				if (count == CONNECTION_FACTOR) {
					break;
				}
			}
			
			if (!this.nodes.contains(initialNodeId)) {
				this.nodes.add(initialNodeId.toString());
			}			
		}
		
		logger.info("group " + this.groupId + " connected");
		
		logger.info("start balancing ...");
		Message messageNodeAgentsLoadRequest = MessageFactory.NodeAgentsLoadRequest(connectionManager.getNodeId().toString());
		connectionManager.getGroupCommunication().broadcast(messageNodeAgentsLoadRequest);
		
		try {
			Thread.sleep(Configuration.BROADCAST_WAIT);
		} catch (InterruptedException e) {
			logger.error("error while waiting for node agent loads response: " + e.getMessage());
		}
		
		Map<String, Integer> agentsLoad = MessageProcessor.instance.getAgentsLoad();
		int avg = agentsLoad.size() == 0 ? getTotalAgents(agentsLoad): Math.round(((float)getTotalAgents(agentsLoad)) / (agentsLoad.size()));
		// migratelist is the list which i have to take agent from
		Map<String, Integer> migrateList = getMigrateList(agentsLoad, avg);
		
		Queue<AgentDescriptor> listAgents = getListAgents(migrateList, avg);		
		balance(agentsLoad, listAgents, avg, false);
		logger.info("group balanced");
	}

	private Queue<AgentDescriptor> getListAgents(Map<String, Integer> migrateList, int avg) throws RemoteException {
		Queue<AgentDescriptor> listAgents = new LinkedBlockingQueue<AgentDescriptor>();
		Iterator<String> it = migrateList.keySet().iterator();
		while (it.hasNext()) {
			String node = it.next();
			Collection<AgentDescriptor> list = connectionManager.getConnectionManager(node)
													.getSimulationCommunication()
													.migrateAgents(migrateList.get(node) - avg);
			listAgents.addAll(list);
		}
		return listAgents;
	}

	private void balance(Map<String, Integer> agentsLoad, Queue<AgentDescriptor> listAgents, int avg, boolean disconnect) throws RemoteException {
		Map<String, Integer> migrateList = getMigrateList(agentsLoad, avg);
		agentsLoad.keySet().removeAll(migrateList.keySet());
		
		if (disconnect && agentsLoad.size() == 0) {
			// disconnect the last node
			return;
		}
		
		synchronized (listAgents) {
			// lastNode used when after balanced agents, there is a agent left, then added to lastNode
			String lastNode = null;
			for (String n: agentsLoad.keySet()) {
				lastNode = n;
				int agentCountToTransfer = avg - agentsLoad.get(n);
				for (int i = 0; i < agentCountToTransfer; i++) {
					if (listAgents.isEmpty()) {
						MessageProcessor.instance.resetLoads();
						return;
					}
					connectionManager.getConnectionManager(n).getSimulationCommunication().startAgent(listAgents.poll());
				}
			}
			
			// when there left a single agent on the queue
			if (!listAgents.isEmpty()) {
				if (disconnect) {
					logger.debug("add last agent to: " + lastNode);
					connectionManager.getConnectionManager(lastNode).getSimulationCommunication().startAgent(listAgents.poll());
				} else {
					connectionManager.getSimulationCommunication().startAgent(listAgents.poll());
				}
			}
			
			if (!listAgents.isEmpty()) {
				logger.error("error balancing agents: agent missed");
			}
		}
		
		
		MessageProcessor.instance.resetLoads();
	}

	private Map<String, Integer> getMigrateList(Map<String, Integer> nodesLoad,
			int avg) {
		Map<String, Integer> ret = new HashMap<String, Integer>();
		
		for (String node: nodesLoad.keySet()) {
			int loads = nodesLoad.get(node);
			if (loads > avg) {
				ret.put(node, loads);
			}
		}
		
		return ret;
	}

	private int getTotalAgents(Map<String, Integer> ret) {
		int result = 0;
		Iterator<Integer> it = ret.values().iterator();
		while (it.hasNext()) {
			result += it.next();
		}
		return result;
	}

	@Override
	public void createGroup() throws RemoteException {
		this.groupId = Long.toString(System.currentTimeMillis());
		logger.info("group created: " + this.groupId);
	}

	@Override
	public void disconnectFromGroup(String nodeId) throws RemoteException {
		NodeId node = new NodeId(nodeId);
		logger.info("start the migration and balancing process of the disconnected node: " + nodeId + " ... ");
		Message messageNodeAgentsLoadRequest = MessageFactory.NodeAgentsLoadRequest(connectionManager.getNodeId().toString());
		connectionManager.getGroupCommunication().broadcast(messageNodeAgentsLoadRequest);
		
		try {
			Thread.sleep(Configuration.BROADCAST_WAIT);
		} catch (InterruptedException e) {
			logger.error("error while waiting for node agent loads response: " + e.getMessage());
		}
		
		ConcurrentHashMap<String, Integer> agentsLoad = MessageProcessor.instance.getAgentsLoad();		
		Queue<AgentDescriptor> listAgentOfDisconnectNode = new LinkedBlockingQueue<AgentDescriptor>();
		
		// descometar dps esta linea!~!!!
		int disconnectNodeAgentLoad = agentsLoad.get(nodeId);
		//while (agentsLoad.size() == 0) {
			//try {
				//Thread.sleep(Configuration.BROADCAST_WAIT);
//			} catch (InterruptedException e) {
	//			logger.error("error while waiting for node agent loads response: " + e.getMessage());
		//	}
		//}
		

		
		if (disconnectNodeAgentLoad > 0) {
			// get all agent of a disconnected node
			listAgentOfDisconnectNode.addAll(connectionManager
					.getConnectionManager(nodeId)
					.getSimulationCommunication()
					.migrateAgents(disconnectNodeAgentLoad ));
			
			agentsLoad.remove(node.toString());
			
			int avg;
			if (agentsLoad.size() == 0) {
				avg = getTotalAgents(agentsLoad) + disconnectNodeAgentLoad;
			} else {
				avg = Math.round(((float)(getTotalAgents(agentsLoad) + disconnectNodeAgentLoad)) / agentsLoad.size());
			}
			
			balance(agentsLoad, listAgentOfDisconnectNode, avg, true);
			
			logger.info("group balanced");
		}
		
		Message message = MessageFactory.Disconnect(connectionManager.getNodeId().toString(), node.toString());
		connectionManager.getGroupCommunication().broadcast(message);
	}

	@Override
	public String getGroupId() throws RemoteException {
		return this.groupId;
	}

	@Override
	public boolean isConnectedToGroup() throws RemoteException {
		return this.groupId != null;
	}
	
	public Iterator<String> getConnectedNodes() {
		return this.nodes.iterator();
	}

	public void disconnect() {
		synchronized (nodes) {
			nodes.clear();
			
		}
		groupId = null;
		
	}

	public void removeNode(String node) throws RemoteException {
		synchronized (nodes) {
			nodes.remove(node);
		}
		
		String myNodeId = connectionManager.getNodeId().toString();		

//		for (String n: nodes) {			
//			connectionManager.getConnectionManager(n).getClusterAdmimnistration().connectToGroup(myNodeId);
//		}
		
		Iterator<String> it = nodes.iterator();
		while (it.hasNext()) {
			connectionManager.getConnectionManager(it.next()).getClusterAdmimnistration().addNewNode(myNodeId);
		}
	}
	
	public void eliminateNodes(List<String> eliminateNodes) {
		for (String n: eliminateNodes) {
			nodes.remove(n);
		}
	}

}
